package com.bsj.sipgateway.mq.consumer;

import com.alibaba.fastjson.JSON;
import com.bsj.power.common.config.rocketmq.constant.ConsumerGroupConstant;
import com.bsj.power.common.constant.MqTopicAndTagConstants;
import com.bsj.power.common.def.dto.media.BackVodParam;
import com.bsj.power.common.def.dto.video.BackVodControl;
import com.bsj.power.common.def.dto.video.CmdToMqDto;
import com.bsj.power.common.def.entity.gb28181.Device;
import com.bsj.sipgateway.cache.DeviceCache;
import com.bsj.sipgateway.command.RocketMqCommand;
import com.bsj.sipgateway.common.constant.Constants;
import com.bsj.sipgateway.mq.producer.CommandUpProducer;
import com.bsj.sipgateway.sip.session.VideoStreamSessionManager;
import com.bsj.sipgateway.sip.transmit.cmd.impl.SIPCommander;
import com.bsj.sipgateway.util.CommandDownType;
import com.bsj.sipgateway.web.service.impl.DeviceServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

/**
 * @Description 视频mq消费端，主要消费decode发来的视频指令
 * @Author zhaoyang
 * @Date 2024/6/14 19:17
 */
@Slf4j
@Component
@RocketMQMessageListener(topic = MqTopicAndTagConstants.MEDIA_SERVER_CHANGE_TOPIC,
        consumerGroup = ConsumerGroupConstant.TRAVEL_VIDEO_DOWN_MSG_GROUP,
        selectorExpression = MqTopicAndTagConstants.MEDIA_LIVE_STREAM_OPEN_TAG + " || " + MqTopicAndTagConstants.MEDIA_LIVE_BACK_OPEN_TAG,
        messageModel = MessageModel.BROADCASTING)
public class MqttVideoRocketmqConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private DeviceServiceImpl deviceService;

    @Autowired
    private SIPCommander sipCommander;

    @Autowired
    private VideoStreamSessionManager videoStreamSessionManager;

    @Autowired
    private CommandUpProducer commandUpProducer;

    @Async("powerVideoAsync")
    @Override
    public void onMessage(MessageExt msg) {
        String msgBody = new String(msg.getBody());
        String tags = msg.getTags();
        CmdToMqDto cmdToMqDto = JSON.parseObject(msgBody, CmdToMqDto.class);
//        log.info("视频子系统MQ消息,tag: {}，cmdToMqDto：{}", tags, cmdToMqDto.toString());
        String terminalNo = cmdToMqDto.getTerminalNo();
        Long did = cmdToMqDto.getVehicleId();
        Integer channelSerial = cmdToMqDto.getChannelNo();
        Integer cmd = cmdToMqDto.getCmd();
        Integer control = cmdToMqDto.getControl();
        String resolutionRatio = cmdToMqDto.getResolutionRatio();
        Integer bitStream = null;
        // 先根据did查询数据库有没有设备
        Device device = deviceService.selectById(did);
        if (device == null) {
//            log.warn("接收到mq消息，数据库没有这台设备，设备数据库id={}", did);
            return;
        }
        // 再判断设备有没有在线，如果不在线就不处理
//        Device deviceInCache = DeviceCache.getDevice(device.getDeviceId());
//        if (deviceInCache == null) {
//            log.warn("接收到mq指令，设备不在线，取消下发mq指令，deviceId={}", device.getDeviceId());
//            return;
//        }
        String deviceId = device.getDeviceId();

        switch (tags) {
            case MqTopicAndTagConstants.MEDIA_LIVE_STREAM_OPEN_TAG: // 直播指令
                if (cmd == 0) {      // 0是直播
                    if (control == 0) { // 0是开始
                        if (!StringUtils.isEmpty(resolutionRatio)) {
                            bitStream = Integer.parseInt(resolutionRatio);
                        }
                        sipCommander.playStreamCmd(device.getDeviceId(), channelSerial, bitStream);
//                        log.info("已发送打开直播指令，did={},deivceId={},channelSerial={}", did, deviceId, channelSerial);
                    } else if (control == 1) {
                        sipCommander.streamByeCmd(deviceId, cmd + "", channelSerial);
//                        log.info("已发送关闭直播指令，did={},deivceId={},channelSerial={}", did, deviceId, channelSerial);
                    }
                }
                break;
            case MqTopicAndTagConstants.MEDIA_LIVE_BACK_OPEN_TAG: // 回放指令
                if (cmd == 1) { // 1是回放
                    if (control == 0) { // 0是开始
                        BackVodParam backVodParam = cmdToMqDto.getBackVodParam(); // 视频回放开始请求参数
                        if (backVodParam == null || StringUtils.isEmpty(backVodParam.getStartTime()) || StringUtils.isEmpty(backVodParam.getEndTime())) {
                            log.warn("接收到mq回放开始指令，backVodParam参数或者事件为空，不下发回放开始指令");
                        } else {
                            long startTime = Long.parseLong(backVodParam.getStartTime());
                            long endTime = Long.parseLong(backVodParam.getEndTime());
                            Integer port = backVodParam.getPort();
                            sipCommander.playbackStreamCmd(startTime, endTime, deviceId, channelSerial, port);
                            log.info("接收到mq回放开始指令，下发回放指令");
                        }
                    } else if (control == 2) { // 2是控制
                        BackVodControl backVodControl = cmdToMqDto.getBackVodControl();
                        Integer controlType = backVodControl.getControlType(); // 这个是回放的控制
                        if (controlType == 2) { // 这个值是2表示关闭回放
                            sipCommander.streamByeCmd(device.getDeviceId(), cmd + "", backVodControl.getChannelNo());
                        } else if (controlType == 0) { // 这个表示暂停后恢复
                            sipCommander.playResumeCmd(did, backVodControl.getChannelNo());
                        } else if (controlType == 1) { // 这个表示暂停
                            sipCommander.playPauseCmd(did, backVodControl.getChannelNo());
                        } else if (controlType == 3) { // 这个表示倍速播放
                            Integer speed = backVodControl.getSpeed();
                            Double d = null;
                            if (speed == 6) { // 四分之一倍速
                                d = 0.25D;
                            } else if (speed == 7) {
                                d = 0.5D;
                            } else if (speed == 1) {
                                d = 1.0D;
                            } else if (speed == 2) {
                                d = 2.0D;
                            } else if (speed == 4) {
                                d = 4.0D;
                            }
                            sipCommander.playSpeedCmd(did, channelSerial, d);

                        }
                        log.info("已发送关闭回放指令，did={},deivceId={},channelSerial={}", did, deviceId, backVodControl.getChannelNo());
                    }
                }
                break;
            default:
                log.warn("暂不支持处理的tag类型，tag={}", tags);
                break;
        }


    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        // 设置消费者消息重试次数
        defaultMQPushConsumer.setMaxReconsumeTimes(3);
        //设置实例名称
        defaultMQPushConsumer.setInstanceName(MqTopicAndTagConstants.MQTT_PROCESS_VIDEO_INSTANCE_NAME);
    }


    public void deal(RocketMqCommand command) {
        Device device = DeviceCache.getDevice(command.getDevice());
        if (device == null) {
            //设备不在线
            command.setStatus(0);
            command.setRetMsg("设备不在线");
        } else {
            switch (command.getCmdType()) {
                case CommandDownType.REAL_MEDIA_OPEN_CMD:
                case CommandDownType.REAL_AUDIO_OPEN_CMD:
                case CommandDownType.REAL_VIDEO_OPEN_CMD:
                    //打开音视频或视频
//                    sipCommander.playStreamCmd(command, command.getDevice());
                    break;
                case CommandDownType.DEV_VOD_MEDIA_QUERY:
                    //设备回放查询指令
                    String cmdMessage = command.getCmdMessage();
                    if (StringUtils.isNotBlank(cmdMessage) && cmdMessage.length() >= 26) {
                        String startTimeBcd = cmdMessage.substring(2, 14);
                        String endTimeBcd = cmdMessage.substring(14, 26);
                        log.info("设备{}回放查询，开始时间：{}，结束时间{}", command.getDevice(), startTimeBcd, endTimeBcd);

//                        sipCommander.recordInfoQuery(command, startTimeBcd, endTimeBcd);
                    } else {
                        command.setStatus(0);
                        command.setRetMsg("设备回放查询指令参数错误");
                        log.error("设备回放查询指令参数错误，{}", JSON.toJSONString(command));
                    }
                    break;
                case CommandDownType.DEV_VOD_MEDIA_TRANS:
                    //设备回放传输指令
//                    sipCommander.playbackStreamCmd(command, command.getDevice());
                    break;
                case CommandDownType.REAL_INTERCOM_OPEN_CMD:
                    //打开对讲
                    if (videoStreamSessionManager.contain(command.getDevice(), Constants.BROADCAST_PREFIX, command.getSpareMessage())) {
                        //已经打开对讲，mq响应失败
                        command.setStatus(0);
                        command.setRetMsg("设备正在对讲中");
                    } else {
                        //先下发实时视频点播，等ACK后再下发语音广播
//                        sipCommander.playStreamCmd(command, command.getDevice());
                    }
                    break;
                case CommandDownType.REAL_MEDIA_CTRL_CLOSEALL:
                case CommandDownType.REAL_MEDIA_CTRL_CLOSEINTERCOM:
                    //关闭实时视频
//                    sipCommander.streamByeCmd(device.getDeviceId(), Constants.PLAYER_PREFIX, command);
                    break;
                case CommandDownType.DEV_VOD_MEDIA_CLOSE:
                    //关闭历史视频
//                    sipCommander.streamByeCmd(device.getDeviceId(), Constants.PLAY_BLACK_PREFIX, command);
                    break;
                case CommandDownType.PTZ_ROTATE:
                case CommandDownType.PTZ_ADJUSTING_FOCAL_CTRL:
                case CommandDownType.PTZ_ADJUSTING_APERTURE_CTRL:
                case CommandDownType.PTZ_WIPER_CTRL:
                case CommandDownType.PTZ_ZOOM_CTRL:
//                    sipCommander.ptz(command, command.getDevice());
                    break;
            }
        }
        //响应mq
        commandUpProducer.send(command, Constants.SQL_INSERT);
    }

}
